package jobs;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import domain.TextPair;

public class JoinProcessor {
	public static final String FIRSTNAME_TYPE = "firstname.csv";
	
	public static class JoinMapper extends
    Mapper<Text, Text, Text, TextPair> {
	@Override
	public void map(Text key, Text value, Context context)
	        throws IOException,InterruptedException {
			String type = ((FileSplit)context.getInputSplit()).getPath().getName();
	        context.write(new Text(key.toString()),new TextPair(value.toString(),type));
		}
	}

	public static class JoinReducer extends Reducer<Text,TextPair,Text,Text>{
        @Override
        public void reduce(Text key , Iterable <TextPair> values , Context context) 
                throws IOException,InterruptedException {
        	for(TextPair val1 : values){
        		TextPair v1 = new TextPair(val1.getFirst().toString(),val1.getSecond().toString());
        		for(TextPair val2 : values){
        			if(!v1.getSecond().equals(val2.getSecond())){
        				if(FIRSTNAME_TYPE.equals(v1.getSecond().toString())){
        					context.write(key, new Text(v1.getFirst() + " - " + val2.getFirst()));
        				}else{
        					context.write(key, new Text(val2.getFirst() + " - " + v1.getFirst()));
        				}
        				
        			}
        		}
        	}
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.set("mapreduce.input.keyvaluelinerecordreader.key.value.separator", "|");
        Job job = new Job(conf, "Join Processor");
        job.setJarByClass(JoinProcessor.class);
        job.setMapperClass(JoinMapper.class);
        job.setReducerClass(JoinReducer.class);
        job.setPartitionerClass(JoinPartitioner.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(TextPair.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        job.setInputFormatClass(JoinInputFormat.class);
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}